/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client;

import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.utils.NameServerAddressUtils;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RequestType;

/**
 * Client Common configuration
 */
public class ClientConfig {
    //1、控制是否通过 VIP 通道发送消息，默认为 false（不开启）。
    public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
    //2、配置使用 SOCKS 代理发送消息。示例中通过系统属性获取配置值，并设定了默认值为 {}。
    public static final String SOCKS_PROXY_CONFIG = "com.rocketmq.socks.proxy.config";
    //3、控制是否解码消息体，默认为 true（开启）。
    public static final String DECODE_READ_BODY = "com.rocketmq.read.body";
    //4、控制是否解码压缩的消息体，默认为 true（开启）。
    public static final String DECODE_DECOMPRESS_BODY = "com.rocketmq.decompress.body";
    //控制是否开启消息发送延迟统计功能，默认为 false（不开启）。当启用时，RocketMQ 会记录每次消息发送的延迟时间。
    //5、sendLatencyEnable 是 RocketMQ 中 Producer 端的一个配置项，用于控制是否开启消息发送延迟统计功能。当启用此功能时，RocketMQ 会记录每次消息发送的延迟时间（即从消息被发送到消息被成功存储或确认的时间差），这对于监控和性能分析非常有用。
    public static final String SEND_LATENCY_ENABLE = "com.rocketmq.sendLatencyEnable";
    //6、控制是否启动探测器，默认为 false（不开启）。
    public static final String START_DETECTOR_ENABLE = "com.rocketmq.startDetectorEnable";
    //7、控制是否使用心跳版本2，默认为 false（不开启）。
    public static final String HEART_BEAT_V2 = "com.rocketmq.heartbeat.v2";
    //8、获取 NameServer 地址。如果未设置，则使用默认的 NameServer 地址。
    private String namesrvAddr = NameServerAddressUtils.getNameServerAddresses();
    //9、客户端 IP 地址，默认为本地地址。
    private String clientIP = NetworkUtil.getLocalAddress();
    //10、客户端名称，默认为 "DEFAULT"。
    private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
    //11、客户端回调执行线程数，默认为可用处理器数量。
    private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
    @Deprecated
    protected String namespace;
    //12、是否已经初始化命名空间，默认为 false（未初始化）。
    private boolean namespaceInitialized = false;
    //13、命名空间版本，默认为空字符串。
    protected String namespaceV2;
    //14、
    protected AccessChannel accessChannel = AccessChannel.LOCAL;

    /**
     * Pulling topic information interval from the named server
     * 从 NameServer 拉取主题信息的时间间隔，默认为 30秒
     */
    private int pollNameServerInterval = 1000 * 30;
    /**
     * Heartbeat interval in microseconds with message broker
     * 向消息代理发送心跳的时间间隔，默认为 30 微秒；百万分之一秒。
     */
    private int heartbeatBrokerInterval = 1000 * 30;
    /**
     * Offset persistent interval for consumer
     * 消费者偏移量持久化的时间间隔，默认为 5秒
     */
    private int persistConsumerOffsetInterval = 1000 * 5;
    //15、发生异常时拉取消息的延迟时间，默认为 1秒
    private long pullTimeDelayMillsWhenException = 1000;
    //16、单元模式，默认为 false。
    private boolean unitMode = false;
    //17、单元名称，默认为空字符串。
    private String unitName;
    //控制是否解码消息体，默认为 true（开启）。
    private boolean decodeReadBody = Boolean.parseBoolean(System.getProperty(DECODE_READ_BODY, "true"));
    //控制是否解码压缩的消息体，默认为 true（开启）。
    private boolean decodeDecompressBody = Boolean.parseBoolean(System.getProperty(DECODE_DECOMPRESS_BODY, "true"));
    //18、是否通过 VIP 通道发送消息，默认为 false（不开启）。
    private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "false"));
    //19、是否使用心跳版本2，默认为 false（不开启）。
    private boolean useHeartbeatV2 = Boolean.parseBoolean(System.getProperty(HEART_BEAT_V2, "false"));
    //20、是否开启 TLS 加密通信，默认为 TlsSystemConfig.tlsEnable 的值。
    //TLS（Transport Layer Security，传输层安全协议）是一种加密协议，设计用于提供通信安全性和数据完整性。
    private boolean useTLS = TlsSystemConfig.tlsEnable;
    //21、配置使用 SOCKS 代理发送消息。示例中通过系统属性获取配置值，并设定了默认值为 {}。
    private String socksProxyConfig = System.getProperty(SOCKS_PROXY_CONFIG, "{}");
    //22、客户端 API 超时时间，默认为 3秒。
    private int mqClientApiTimeout = 3 * 1000;
    //23、检测超时时间，默认为 200毫秒。
    private int detectTimeout = 200;
    //24、检测间隔时间，默认为 2秒。
    private int detectInterval = 2 * 1000;
    //25、语言代码，默认为 JAVA。
    private LanguageCode language = LanguageCode.JAVA;

    /**
     * Enable stream request type will inject a RPCHook to add corresponding request type to remoting layer.
     * And it will also generate a different client id to prevent unexpected reuses of MQClientInstance.
     * 启用流请求类型将向远程层注入一个 RPCHook，以添加相应的请求类型。 同时，它也会生成一个不同的客户端ID以防止MQClientInstance的意外重用。
     */
    protected boolean enableStreamRequestType = false;

    /**
     * Enable the fault tolerance mechanism of the client sending process.
     * DO NOT OPEN when ORDER messages are required.
     * Turning on will interfere with the queue selection functionality,
     * possibly conflicting with the order message.
     * 启用客户端发送过程的容错机制。需要顺序消息的情况下不要打开。开启后将干扰队列选择功能，可能和顺序消息冲突。
     */
    private boolean sendLatencyEnable = Boolean.parseBoolean(System.getProperty(SEND_LATENCY_ENABLE, "false"));
    //26、控制是否启动探测器，默认为 false（不开启）。
    private boolean startDetectorEnable = Boolean.parseBoolean(System.getProperty(START_DETECTOR_ENABLE, "false"));

    private boolean enableHeartbeatChannelEventListener = true;

    /**
     * 构建 MQClientId，
     * the result is like: ip@client-uniqueString[pid]
     * @return
     */
    public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClientIP());

        sb.append("@");
        sb.append(this.getInstanceName());
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }

        if (enableStreamRequestType) {
            sb.append("@");
            sb.append(RequestType.STREAM);
        }

        return sb.toString();
    }

    public String getClientIP() {
        return clientIP;
    }

    public void setClientIP(String clientIP) {
        this.clientIP = clientIP;
    }

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public void changeInstanceNameToPID() {
        if (this.instanceName.equals("DEFAULT")) {
            this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
        }
    }

    @Deprecated
    public String withNamespace(String resource) {
        return NamespaceUtil.wrapNamespace(this.getNamespace(), resource);
    }

    @Deprecated
    public Set<String> withNamespace(Set<String> resourceSet) {
        Set<String> resourceWithNamespace = new HashSet<>();
        for (String resource : resourceSet) {
            resourceWithNamespace.add(withNamespace(resource));
        }
        return resourceWithNamespace;
    }

    @Deprecated
    public String withoutNamespace(String resource) {
        return NamespaceUtil.withoutNamespace(resource, this.getNamespace());
    }

    @Deprecated
    public Set<String> withoutNamespace(Set<String> resourceSet) {
        Set<String> resourceWithoutNamespace = new HashSet<>();
        for (String resource : resourceSet) {
            resourceWithoutNamespace.add(withoutNamespace(resource));
        }
        return resourceWithoutNamespace;
    }

    @Deprecated
    public MessageQueue queueWithNamespace(MessageQueue queue) {
        if (StringUtils.isEmpty(this.getNamespace())) {
            return queue;
        }
        return new MessageQueue(withNamespace(queue.getTopic()), queue.getBrokerName(), queue.getQueueId());
    }

    @Deprecated
    public Collection<MessageQueue> queuesWithNamespace(Collection<MessageQueue> queues) {
        if (StringUtils.isEmpty(this.getNamespace())) {
            return queues;
        }
        Iterator<MessageQueue> iter = queues.iterator();
        while (iter.hasNext()) {
            MessageQueue queue = iter.next();
            queue.setTopic(withNamespace(queue.getTopic()));
        }
        return queues;
    }

    public void resetClientConfig(final ClientConfig cc) {
        this.namesrvAddr = cc.namesrvAddr;
        this.clientIP = cc.clientIP;
        this.instanceName = cc.instanceName;
        this.clientCallbackExecutorThreads = cc.clientCallbackExecutorThreads;
        this.pollNameServerInterval = cc.pollNameServerInterval;
        this.heartbeatBrokerInterval = cc.heartbeatBrokerInterval;
        this.persistConsumerOffsetInterval = cc.persistConsumerOffsetInterval;
        this.pullTimeDelayMillsWhenException = cc.pullTimeDelayMillsWhenException;
        this.unitMode = cc.unitMode;
        this.unitName = cc.unitName;
        this.vipChannelEnabled = cc.vipChannelEnabled;
        this.useTLS = cc.useTLS;
        this.socksProxyConfig = cc.socksProxyConfig;
        this.namespace = cc.namespace;
        this.language = cc.language;
        this.mqClientApiTimeout = cc.mqClientApiTimeout;
        this.decodeReadBody = cc.decodeReadBody;
        this.decodeDecompressBody = cc.decodeDecompressBody;
        this.enableStreamRequestType = cc.enableStreamRequestType;
        this.useHeartbeatV2 = cc.useHeartbeatV2;
        this.startDetectorEnable = cc.startDetectorEnable;
        this.sendLatencyEnable = cc.sendLatencyEnable;
        this.enableHeartbeatChannelEventListener = cc.enableHeartbeatChannelEventListener;
        this.detectInterval = cc.detectInterval;
        this.detectTimeout = cc.detectTimeout;
        this.namespaceV2 = cc.namespaceV2;
    }

    public ClientConfig cloneClientConfig() {
        ClientConfig cc = new ClientConfig();
        cc.namesrvAddr = namesrvAddr;
        cc.clientIP = clientIP;
        cc.instanceName = instanceName;
        cc.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
        cc.pollNameServerInterval = pollNameServerInterval;
        cc.heartbeatBrokerInterval = heartbeatBrokerInterval;
        cc.persistConsumerOffsetInterval = persistConsumerOffsetInterval;
        cc.pullTimeDelayMillsWhenException = pullTimeDelayMillsWhenException;
        cc.unitMode = unitMode;
        cc.unitName = unitName;
        cc.vipChannelEnabled = vipChannelEnabled;
        cc.useTLS = useTLS;
        cc.socksProxyConfig = socksProxyConfig;
        cc.namespace = namespace;
        cc.language = language;
        cc.mqClientApiTimeout = mqClientApiTimeout;
        cc.decodeReadBody = decodeReadBody;
        cc.decodeDecompressBody = decodeDecompressBody;
        cc.enableStreamRequestType = enableStreamRequestType;
        cc.useHeartbeatV2 = useHeartbeatV2;
        cc.startDetectorEnable = startDetectorEnable;
        cc.enableHeartbeatChannelEventListener = enableHeartbeatChannelEventListener;
        cc.sendLatencyEnable = sendLatencyEnable;
        cc.detectInterval = detectInterval;
        cc.detectTimeout = detectTimeout;
        cc.namespaceV2 = namespaceV2;
        return cc;
    }

    public String getNamesrvAddr() {
        if (StringUtils.isNotEmpty(namesrvAddr) && NameServerAddressUtils.NAMESRV_ENDPOINT_PATTERN.matcher(namesrvAddr.trim()).matches()) {
            return NameServerAddressUtils.getNameSrvAddrFromNamesrvEndpoint(namesrvAddr);
        }
        return namesrvAddr;
    }

    /**
     * Domain name mode access way does not support the delimiter(;), and only one domain name can be set.
     *
     * @param namesrvAddr name server address
     */
    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
        this.namespaceInitialized = false;
    }

    public int getClientCallbackExecutorThreads() {
        return clientCallbackExecutorThreads;
    }

    public void setClientCallbackExecutorThreads(int clientCallbackExecutorThreads) {
        this.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
    }

    public int getPollNameServerInterval() {
        return pollNameServerInterval;
    }

    public void setPollNameServerInterval(int pollNameServerInterval) {
        this.pollNameServerInterval = pollNameServerInterval;
    }

    public int getHeartbeatBrokerInterval() {
        return heartbeatBrokerInterval;
    }

    public void setHeartbeatBrokerInterval(int heartbeatBrokerInterval) {
        this.heartbeatBrokerInterval = heartbeatBrokerInterval;
    }

    public int getPersistConsumerOffsetInterval() {
        return persistConsumerOffsetInterval;
    }

    public void setPersistConsumerOffsetInterval(int persistConsumerOffsetInterval) {
        this.persistConsumerOffsetInterval = persistConsumerOffsetInterval;
    }

    public long getPullTimeDelayMillsWhenException() {
        return pullTimeDelayMillsWhenException;
    }

    public void setPullTimeDelayMillsWhenException(long pullTimeDelayMillsWhenException) {
        this.pullTimeDelayMillsWhenException = pullTimeDelayMillsWhenException;
    }

    public String getUnitName() {
        return unitName;
    }

    public void setUnitName(String unitName) {
        this.unitName = unitName;
    }

    public boolean isUnitMode() {
        return unitMode;
    }

    public void setUnitMode(boolean unitMode) {
        this.unitMode = unitMode;
    }

    public boolean isVipChannelEnabled() {
        return vipChannelEnabled;
    }

    public void setVipChannelEnabled(final boolean vipChannelEnabled) {
        this.vipChannelEnabled = vipChannelEnabled;
    }

    public boolean isUseTLS() {
        return useTLS;
    }

    public void setUseTLS(boolean useTLS) {
        this.useTLS = useTLS;
    }

    public String getSocksProxyConfig() {
        return socksProxyConfig;
    }

    public void setSocksProxyConfig(String socksProxyConfig) {
        this.socksProxyConfig = socksProxyConfig;
    }

    public LanguageCode getLanguage() {
        return language;
    }

    public void setLanguage(LanguageCode language) {
        this.language = language;
    }

    public boolean isDecodeReadBody() {
        return decodeReadBody;
    }

    public void setDecodeReadBody(boolean decodeReadBody) {
        this.decodeReadBody = decodeReadBody;
    }

    public boolean isDecodeDecompressBody() {
        return decodeDecompressBody;
    }

    public void setDecodeDecompressBody(boolean decodeDecompressBody) {
        this.decodeDecompressBody = decodeDecompressBody;
    }

    @Deprecated
    public String getNamespace() {
        if (namespaceInitialized) {
            return namespace;
        }

        if (StringUtils.isNotEmpty(namespace)) {
            return namespace;
        }

        if (StringUtils.isNotEmpty(this.namesrvAddr)) {
            if (NameServerAddressUtils.validateInstanceEndpoint(namesrvAddr)) {
                namespace = NameServerAddressUtils.parseInstanceIdFromEndpoint(namesrvAddr);
            }
        }
        namespaceInitialized = true;
        return namespace;
    }

    @Deprecated
    public void setNamespace(String namespace) {
        this.namespace = namespace;
        this.namespaceInitialized = true;
    }

    public String getNamespaceV2() {
        return namespaceV2;
    }

    public void setNamespaceV2(String namespaceV2) {
        this.namespaceV2 = namespaceV2;
    }

    public AccessChannel getAccessChannel() {
        return this.accessChannel;
    }

    public void setAccessChannel(AccessChannel accessChannel) {
        this.accessChannel = accessChannel;
    }

    public int getMqClientApiTimeout() {
        return mqClientApiTimeout;
    }

    public void setMqClientApiTimeout(int mqClientApiTimeout) {
        this.mqClientApiTimeout = mqClientApiTimeout;
    }

    public boolean isEnableStreamRequestType() {
        return enableStreamRequestType;
    }

    public void setEnableStreamRequestType(boolean enableStreamRequestType) {
        this.enableStreamRequestType = enableStreamRequestType;
    }

    public boolean isSendLatencyEnable() {
        return sendLatencyEnable;
    }

    public void setSendLatencyEnable(boolean sendLatencyEnable) {
        this.sendLatencyEnable = sendLatencyEnable;
    }

    public boolean isStartDetectorEnable() {
        return startDetectorEnable;
    }

    public void setStartDetectorEnable(boolean startDetectorEnable) {
        this.startDetectorEnable = startDetectorEnable;
    }

    public boolean isEnableHeartbeatChannelEventListener() {
        return enableHeartbeatChannelEventListener;
    }

    public void setEnableHeartbeatChannelEventListener(boolean enableHeartbeatChannelEventListener) {
        this.enableHeartbeatChannelEventListener = enableHeartbeatChannelEventListener;
    }

    public int getDetectTimeout() {
        return this.detectTimeout;
    }

    public void setDetectTimeout(int detectTimeout) {
        this.detectTimeout = detectTimeout;
    }

    public int getDetectInterval() {
        return this.detectInterval;
    }

    public void setDetectInterval(int detectInterval) {
        this.detectInterval = detectInterval;
    }

    public boolean isUseHeartbeatV2() {
        return useHeartbeatV2;
    }

    public void setUseHeartbeatV2(boolean useHeartbeatV2) {
        this.useHeartbeatV2 = useHeartbeatV2;
    }

    @Override
    public String toString() {
        return "ClientConfig{" +
            "namesrvAddr='" + namesrvAddr + '\'' +
            ", clientIP='" + clientIP + '\'' +
            ", instanceName='" + instanceName + '\'' +
            ", clientCallbackExecutorThreads=" + clientCallbackExecutorThreads +
            ", namespace='" + namespace + '\'' +
            ", namespaceInitialized=" + namespaceInitialized +
            ", namespaceV2='" + namespaceV2 + '\'' +
            ", accessChannel=" + accessChannel +
            ", pollNameServerInterval=" + pollNameServerInterval +
            ", heartbeatBrokerInterval=" + heartbeatBrokerInterval +
            ", persistConsumerOffsetInterval=" + persistConsumerOffsetInterval +
            ", pullTimeDelayMillsWhenException=" + pullTimeDelayMillsWhenException +
            ", unitMode=" + unitMode +
            ", unitName='" + unitName + '\'' +
            ", decodeReadBody=" + decodeReadBody +
            ", decodeDecompressBody=" + decodeDecompressBody +
            ", vipChannelEnabled=" + vipChannelEnabled +
            ", useHeartbeatV2=" + useHeartbeatV2 +
            ", useTLS=" + useTLS +
            ", socksProxyConfig='" + socksProxyConfig + '\'' +
            ", mqClientApiTimeout=" + mqClientApiTimeout +
            ", detectTimeout=" + detectTimeout +
            ", detectInterval=" + detectInterval +
            ", language=" + language +
            ", enableStreamRequestType=" + enableStreamRequestType +
            ", sendLatencyEnable=" + sendLatencyEnable +
            ", startDetectorEnable=" + startDetectorEnable +
            ", enableHeartbeatChannelEventListener=" + enableHeartbeatChannelEventListener +
            '}';
    }
}
